package shuai7boy.vip.transform.mr.nu;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Reducer;

import shuai7boy.vip.common.KpiTypeEnum;
import shuai7boy.vip.transformer.model.dim.StatsUserDimension;
import shuai7boy.vip.transformer.model.value.map.TimeOutValue;
import shuai7boy.vip.transformer.model.value.reduce.MapWritableValue;

public class NewUserReducer extends Reducer<StatsUserDimension, TimeOutValue, StatsUserDimension, MapWritableValue> {

	MapWritableValue mapWritableValue = new MapWritableValue();
	MapWritable mapWritable = new MapWritable();
	Set<String> idList = new HashSet<String>();

	protected void reduce(StatsUserDimension key, Iterable<TimeOutValue> values, Context context)
			throws IOException, InterruptedException {

		// 映射维度，KPI，数量
		
		for (TimeOutValue val : values) {
			String userId = val.getId();
			idList.add(userId);
		}
		KpiTypeEnum kpiTypeEnum = KpiTypeEnum.valueOfName(key.getStatsCommon().getKpi().getKpiName());
		mapWritableValue.setKpi(kpiTypeEnum);

		mapWritable.put(new IntWritable(-1), new IntWritable(idList.size()));
		mapWritableValue.setValue(mapWritable);
		context.write(key, mapWritableValue);
	}
}




